package com.ddl.kafka.config.topic;

import com.ddl.kafka.config.MessageData;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "topic_2", durable = "true"),
        exchange = @Exchange(value = "topicExchange", type = ExchangeTypes.TOPIC),
        key = "abc.send.email")
)
@Slf4j
public class TopicConsumer {

    @RabbitHandler
    /*public void onMessage(MessageData message) {
        log.info("[business消费消息][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message.getMsg());
    }*/
    public void onMessage(MessageData message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        // ack 确认消息
        // 第二个参数 multiple ，用于批量确认消息，为了减少网络流量，手动确认可以被批处。
        // 1. 当 multiple 为 true 时，则可以一次性确认 deliveryTag 小于等于传入值的所有消息
        // 2. 当 multiple 为 false 时，则只确认当前 deliveryTag 对应的消息
        channel.basicAck(deliveryTag, true);
        log.info("[business消费消息][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message.getMsg());
    }
}
